package cn.com.ytst.data.integration.core.plugin;


import java.io.UnsupportedEncodingException;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.alibaba.datax.common.util.Configuration;
import cn.com.ytst.data.integration.core.channel.PipleChannel;

/**
 * 每个Writer插件需要实现Writer类，并在其内部实现Job、Task两个内部类。
 */

public abstract class Writer {

    private Logger logger = LoggerFactory.getLogger(Writer.class);

  
    private PipleChannel pipleChannel;

    public abstract void init();

    private Configuration writerConfiguration;

    private AtomicBoolean isstop = new AtomicBoolean(false);


    public AtomicBoolean getIsstop() {
        return isstop;
    }

    private AtomicLong memoryBytes = null;

    private AtomicLong num = null;

    public Writer(){
        memoryBytes = new AtomicLong();
        num = new AtomicLong();
    }

    public void setIsstop(AtomicBoolean isstop) {
        this.isstop = isstop;
    }

    public void writerLine() {
        while (!isstop.get()) {
             String line = this.getPipleChannel().doPull();
            if(null == line){
                isstop.set(true);
                //System.out.println("----------------"+line);
            }else{
                startWriter(line);
                long length = 0;
                try {
                    length = memoryBytes.get() + line.getBytes("utf-8").length;
                    memoryBytes.set(length);
                    num.incrementAndGet();
                } catch (UnsupportedEncodingException e) {
                    e.printStackTrace();
                }

            }
        }
        logger.info("当前共传输字节为{}:共传输数据{}条",memoryBytes.get()/1024,num.get());
        memoryBytes.set(0);
        num.set(0);
    }

    //获取读插件配置
    public void setPluginJobConf(Configuration writerConfiguration) {
        this.writerConfiguration = writerConfiguration;
    }

    public Configuration getPluginJobConf() {
        return this.writerConfiguration;
    }

    public abstract void startWriter(String line);

    public void closed() {
        this.isstop.set(true);
        destroy();
    }

    public abstract void destroy();

    public PipleChannel getPipleChannel() {
        return pipleChannel;
    }

    public void setPipleChannel(PipleChannel pipleChannel) {
        this.pipleChannel = pipleChannel;
    }


}
